/**
 * 
 */
package com.hanhai.common.zookeeper;

import java.io.IOException;
import java.util.List;
import java.util.concurrent.CountDownLatch;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author leiruiqi@hanhai.com
 *
 */

public class BaseZookeeper implements Watcher {

	private static final Logger LOG = LoggerFactory.getLogger(BaseZookeeper.class);  
	   
	public ZooKeeper zooKeeper;

	private static final int SESSION_TIME_OUT = 2000;

	//private CountDownLatch countDownLatch = new CountDownLatch(1);
	
	private String host = "";

	public void setHost(String host) {
		this.host = host;
	}
	public void init() throws IOException, InterruptedException{
		this.connectZookeeper(host);
	}
	/**
	 * 连接zookeeper
	 * 
	 * @param host
	 * @throws IOException
	 * @throws InterruptedException
	 */
	public void connectZookeeper(String host) throws IOException,
			InterruptedException {
		zooKeeper = new ZooKeeper(host, SESSION_TIME_OUT, this);
	//	countDownLatch.await();
		LOG.info("zookeeper connect ok");

	}

	/**
	 * 实现watcher的接口方法，当连接zookeeper成功后，zookeeper会通过此方法通知watcher<br>
	 * 此处为如果接受到连接成功的event，则countDown，让当前线程继续其他事情。
	 */
	@Override
	public void process(WatchedEvent event) {
		Event.KeeperState keeperState = event.getState();  
        Event.EventType eventType = event.getType(); 
        
        if (event.getState() == KeeperState.SyncConnected) {
        	LOG.info("watcher received event");
		//	countDownLatch.countDown();
		} else if ( Event.KeeperState.Disconnected == keeperState ) {  
            LOG.info( "与ZK服务器断开连接" );  
        } else if ( Event.KeeperState.AuthFailed == keeperState ) {  
            LOG.info( "权限检查失败" );  
        } else if ( Event.KeeperState.Expired == keeperState ) {  
            LOG.info( "会话失效" );  
        }  
	}

	/**
	 * 根据路径创建节点，并且设置节点数据
	 * 
	 * @param path
	 * @param data
	 * @return
	 * @throws KeeperException
	 * @throws InterruptedException
	 */
	public String createNode(String path, byte[] data,CreateMode mode) throws KeeperException,
			InterruptedException {
		return this.zooKeeper.create(path, data, Ids.OPEN_ACL_UNSAFE,
				mode);
	}

	/**
	 * 根据路径获取所有孩子节点
	 * 
	 * @param path
	 * @return
	 * @throws KeeperException
	 * @throws InterruptedException
	 */
	public List<String> getChildren(String path) throws KeeperException,
			InterruptedException {
		return this.zooKeeper.getChildren(path, false);
	}

	public Stat setData(String path, byte[] data, int version)
			throws KeeperException, InterruptedException {
		return this.zooKeeper.setData(path, data, version);
	}

	/**
	 * 根据路径获取节点数据
	 * 
	 * @param path
	 * @return
	 * @throws KeeperException
	 * @throws InterruptedException
	 */
	public byte[] getData(String path) throws KeeperException,
			InterruptedException {
		return this.zooKeeper.getData(path, false, null);
	}

	/**
	 * 删除节点
	 * 
	 * @param path
	 * @param version
	 * @throws InterruptedException
	 * @throws KeeperException
	 */
	public void deletNode(String path, int version)
			throws InterruptedException, KeeperException {
		this.zooKeeper.delete(path, version);
	}

	/**
	 * 关闭zookeeper连接
	 * 
	 * @throws InterruptedException
	 */
	public void closeConnect() throws InterruptedException {
		if (null != zooKeeper) {
			zooKeeper.close();
		}
	}

}
